package com.kingsoft.dc.khaos.util

import com.kingsoft.dc.khaos.KhaosConf
import com.kingsoft.dc.khaos.metadata.{Dependency, EdgeInfo, ModuleCommon}
import com.kingsoft.dc.khaos.optimize.{DagOptimize, DependOptimize}
import org.json4s.jackson.JsonMethods.{parse, _}
import org.json4s.{DefaultFormats, _}

import scala.collection.immutable.List
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
  * Created by chengguangqing on 2019/5/31.
  */
class AppConfigParseKsyun(conf: KhaosConf) extends Logging{
  /**
    * 解析程序输入的json信息
    *
    * @param json : 传入json配置信息
    * @return : 返回job的前端配置
    */

  def parseJson(json: String): (String, String, String,Map[String, Any], List[ModuleCommon]) = {

    implicit val formats = DefaultFormats

//    val newJson = json.substring(1,json.length-1).replace("\\\"","\"")
    ////21202F2938212B3E22272626252E434D
    //add 2019/08/30
    //val newJson = json

    val parser = parse(json, true)
    var commonEnv = parser.extract[Map[String, Any]]
    if (!commonEnv.contains("graph")) {
        throw new Exception("配置项graph未设置，请确认作业是否进行了正确的校验转换。")
    }
    val id = try {
      (parser \ "job.id").extract[String]
    } catch {
      case ex: Exception => "9999999"
    }

    val name = try {
      (parser \ "flow.name").extract[String]
    } catch {
      case ex: Exception => "defaultKhaos"
    }



    //val widgets = (parser \ "graph" \ "widgets")
    val widgets = DagOptimize.replaceVirtualModule(json,conf)
    val edges = (parser \ "graph" \ "edges")

    //nodeList 获取所有组件的ID
    val nodeValues = (parser \ "graph" \ "widgets" \ "id").values

    val nodeList = nodeValues.asInstanceOf[List[String]]//.map{id => BigInt(id)}

    //khaosInfo 存放ModuleCommon
    val khaosInfo = new ListBuffer[ModuleCommon]

    //relationList 存放边的关系
    var relationList = List[(EdgeInfo, EdgeInfo)]()

    edges.asInstanceOf[JArray].children.foreach { jv =>
      val source = (jv \ "source").extract[EdgeInfo]
      val destination = (jv \ "destination").extract[EdgeInfo]
      relationList +:= (source, destination)
    }

    //存放出度degreeOut和入度信息degreeIn
    //将data 依赖也弄出来了，留了个口子，没有全部放到moduleCommon中

    //包含etl
    var degreeDataOut = Map[String, List[Dependency]]()
    var degreeDataIn = Map[String, List[Dependency]]()

    var degreeIn = Map[String, List[String]]()
    var degreeOut = Map[String, List[String]]()

    relationList.foreach { r => //(EdgeInfo,EdgeInfo) %soruce,destination%
      val list = degreeDataOut.getOrElse(r._1.node, Nil)
      degreeDataOut += (r._1.node ->
        list.+:(new Dependency(r._1.node, r._1.seq, r._2.node, r._2.seq)))
      val list2 = degreeOut.getOrElse(r._1.node, Nil)
      degreeOut += (r._1.node -> list2.+:(r._2.node))

      val listIn = degreeDataIn.getOrElse(r._2.node, Nil)
      degreeDataIn += (r._2.node ->
        listIn.+:(new Dependency(r._1.node, r._1.seq, r._2.node, r._2.seq)))

      val list3 = degreeIn.getOrElse(r._2.node, Nil)
      degreeIn += (r._2.node -> list3.+:(r._1.node))
    }


    //处理没有关联关系
    nodeList.foreach { node =>
      if (degreeOut.get(node.toString).isEmpty) {
        degreeOut += (node.toString -> Nil)
      }
      if (degreeDataOut.get(node.toString).isEmpty) {
        degreeDataOut += (node.toString -> Nil)
      }
      if (degreeIn.get(node.toString).isEmpty) {
        degreeIn += (node.toString -> Nil)
      }
      if (degreeDataIn.get(node.toString).isEmpty) {
        degreeDataIn += (node.toString -> Nil)
      }
    }

    //最后将ModuleCommon  放入KhaosInfo中 list[ModuleCommon]
    widgets.children.foreach { modConf =>

      val modConfMap = modConf.values.asInstanceOf[Map[String, _]]
      val mid = String.valueOf(modConfMap.get("id").get)
      val mname = String.valueOf(modConfMap.get("name").get)
      val engine = String.valueOf(modConfMap.get("engine").get)
      val clazz = String.valueOf(modConfMap.get("clazz").get)
      val clazzType = String.valueOf(modConfMap.get("type").get)
      val operatorName = String.valueOf(modConfMap.get("operator_name").get)
      val moduleConf = compact(render(modConf)) //输出json字符串
      conf.set(mid, moduleConf)

      //把sourceCos算子中的filter在检核中使用
      val runClass = (modConf \ "strategy" \ "clazz").extract[String]
      if(runClass.contains("source.CosSource")){
        val filter =(modConf \ "strategy" \ "config" \ "filter").extract[String]
        commonEnv = commonEnv.+("source.cos.filter"->filter)
      }

      //add by 20191023
      if("techCheck".equals(operatorName)){
        commonEnv = commonEnv.+("tech.check.tag"->"true")
      }

      //create by 20190731
      DependOptimize.formatSourceSeq(degreeDataIn.getOrElse(mid, Nil))

      val module = new ModuleCommon( //封装一个出度入度
        mid,
        mname,
        engine,
        clazz,
        clazzType,
        moduleConf,
        "",
        degreeIn.getOrElse(mid, Nil),
        degreeDataIn.getOrElse(mid, Nil), //dataDependency
        degreeOut.getOrElse(mid, Nil),
        degreeDataOut.getOrElse(mid,Nil)
      )
      khaosInfo += module
    }
    if(commonEnv.getOrElse("tech.check.tag","tct-null").equals("tct-null")){
        commonEnv = commonEnv.+("tech.check.tag"->"false")
    }

    (id, name,"1", commonEnv,khaosInfo.toList)
  }
}
